package beautiful.butterfly.drds.data_exchange.read_and_write_data.reader.util;


import beautiful.butterfly.drds.data_exchange.error_code.DBUtilErrorCode;
import beautiful.butterfly.drds.data_exchange.error_code.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.read_and_write_data.reader.Constant;
import beautiful.butterfly.drds.data_exchange.read_and_write_data.reader.Key;
import beautiful.butterfly.drds.data_exchange.util.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigInteger;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;


public class SingleTableSplitUtil
{
    private static final Logger LOG = LoggerFactory
            .getLogger(SingleTableSplitUtil.class);


    private SingleTableSplitUtil()
    {
    }

    public static List<Configuration> splitSingleTable(
            Configuration configuration, int adviceNum) throws SQLException
    {
        List<Configuration> pluginParams = new ArrayList<Configuration>();
        List<String> rangeList;
        String splitPkName = configuration.getString(Key.split_key);
        String column = configuration.getString(Key.column);
        String table = configuration.getString(Key.table_name);
        String where = configuration.getString(Key.where, null);
        boolean hasWhere = StringUtils.isNotBlank(where);

        //String splitMode = configuration.getString(Key.SPLIT_MODE, "");
        //if (MysqlConstant.SPLIT_MODE_RANDOMSAMPLE.equals(splitMode) && DATABASE_TYPE == DataBaseType.Oracle) {
        {
            Pair<Object, Object> minMaxPK = getPkRange(configuration);
            if (null == minMaxPK)
            {
                throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK,
                        "根据切分主键切分表失败. DataX 仅支持切分主键为一个,并且类型为整数或者字符串类型. 请尝试使用其他的切分主键或者联系 DBA 进行处理.");
            }

            configuration.set(Key.query_sql, buildQuerySql(column, table, where));
            if (null == minMaxPK.getLeft() || null == minMaxPK.getRight())
            {
                // 切分后获取到的start/end 有 Null 的情况
                pluginParams.add(configuration);
                return pluginParams;
            }

            boolean isStringType = Constant.PK_TYPE_STRING.equals(configuration
                    .getString(Constant.PK_TYPE));
            boolean isLongType = Constant.PK_TYPE_LONG.equals(configuration
                    .getString(Constant.PK_TYPE));


            if (isStringType)
            {
                rangeList = RdbmsRangeSplitWrap.splitAndWrap(
                        String.valueOf(minMaxPK.getLeft()),
                        String.valueOf(minMaxPK.getRight()), adviceNum,
                        splitPkName, "'");
            } else if (isLongType)
            {
                rangeList = RdbmsRangeSplitWrap.splitAndWrap(
                        new BigInteger(minMaxPK.getLeft().toString()),
                        new BigInteger(minMaxPK.getRight().toString()),
                        adviceNum, splitPkName);
            } else
            {
                throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK,
                        "您配置的切分主键(splitPk) 类型 DataX 不支持. DataX 仅支持切分主键为一个,并且类型为整数或者字符串类型. 请尝试使用其他的切分主键或者联系 DBA 进行处理.");
            }
        }
        String tempQuerySql;
        List<String> allQuerySql = new ArrayList<String>();

        if (null != rangeList && !rangeList.isEmpty())
        {
            for (String range : rangeList)
            {
                Configuration tempConfig = configuration.clone();

                tempQuerySql = buildQuerySql(column, table, where)
                        + (hasWhere ? " and " : " where ") + range;

                allQuerySql.add(tempQuerySql);
                tempConfig.set(Key.query_sql, tempQuerySql);
                pluginParams.add(tempConfig);
            }
        } else
        {
            //pluginParams.add(configuration); // this is wrong for new & old doBigIntegerSplit
            Configuration tempConfig = configuration.clone();
            tempQuerySql = buildQuerySql(column, table, where)
                    + (hasWhere ? " and " : " where ")
                    + String.format(" %s IS NOT NULL", splitPkName);
            allQuerySql.add(tempQuerySql);
            tempConfig.set(Key.query_sql, tempQuerySql);
            pluginParams.add(tempConfig);
        }

        // deal pk is null
        Configuration tempConfig = configuration.clone();
        tempQuerySql = buildQuerySql(column, table, where)
                + (hasWhere ? " and " : " where ")
                + String.format(" %s IS NULL", splitPkName);

        allQuerySql.add(tempQuerySql);

        LOG.info("After doBigIntegerSplit(), allQuerySql=[\n{}\n].",
                StringUtils.join(allQuerySql, "\n"));

        tempConfig.set(Key.query_sql, tempQuerySql);
        pluginParams.add(tempConfig);

        return pluginParams;
    }

    public static String buildQuerySql(String column, String table,
                                       String where)
    {
        String querySql;

        if (StringUtils.isBlank(where))
        {
            querySql = String.format(Constant.QUERY_SQL_TEMPLATE_WITHOUT_WHERE,
                    column, table);
        } else
        {
            querySql = String.format(Constant.QUERY_SQL_TEMPLATE, column,
                    table, where);
        }

        return querySql;
    }

    @SuppressWarnings("resource")
    private static Pair<Object, Object> getPkRange(Configuration configuration) throws SQLException
    {
        String pkRangeSQL = genPKRangeSQL(configuration);


        String jdbcURL = configuration.getString(Key.jdbc_url);
        String username = configuration.getString(Key.username);
        String password = configuration.getString(Key.password);
        String table = configuration.getString(Key.table_name);

        Connection conn = Jdbc.getConnectionWithRetry(jdbcURL, username, password);
        Pair<Object, Object> minMaxPK = checkSplitPk(conn, pkRangeSQL, table, username, configuration);
        Jdbc.close(null, null, conn);
        return minMaxPK;
    }

    public static void precheckSplitPk(Connection conn, String pkRangeSQL, int fetchSize,
                                       String table, String username)
    {
        Pair<Object, Object> minMaxPK = checkSplitPk(conn, pkRangeSQL, table, username, null);
        if (null == minMaxPK)
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK,
                    "根据切分主键切分表失败. DataX 仅支持切分主键为一个,并且类型为整数或者字符串类型. 请尝试使用其他的切分主键或者联系 DBA 进行处理.");
        }
    }

    /**
     * 检测splitPk的配置是否正确。
     * configuration为null, 是precheck的逻辑，不需要回写PK_TYPE到configuration中
     */
    private static Pair<Object, Object> checkSplitPk(Connection connection, String pkRangeSQL, String tableName,
                                                     String username, Configuration configuration)
    {

        ResultSet resultSet = null;
        Pair<Object, Object> minMaxPK = null;
        try
        {
            try
            {
                resultSet = Jdbc.executeQuery(connection, pkRangeSQL);
            } catch (Exception e)
            {
                throw RdbmsException.asQueryException(e, pkRangeSQL, tableName, username);
            }
            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
            if (isPKTypeValid(resultSetMetaData))
            {
                if (isStringType(resultSetMetaData.getColumnType(1)))
                {
                    if (configuration != null)
                    {
                        configuration
                                .set(Constant.PK_TYPE, Constant.PK_TYPE_STRING);
                    }
                    while (Jdbcs1.asyncResultSetNext(resultSet))
                    {
                        minMaxPK = new ImmutablePair<Object, Object>(
                                resultSet.getString(1), resultSet.getString(2));
                    }
                } else if (isLongType(resultSetMetaData.getColumnType(1)))
                {
                    if (configuration != null)
                    {
                        configuration.set(Constant.PK_TYPE, Constant.PK_TYPE_LONG);
                    }

                    while (Jdbcs1.asyncResultSetNext(resultSet))
                    {
                        minMaxPK = new ImmutablePair<Object, Object>(
                                resultSet.getString(1), resultSet.getString(2));

                        // check: string shouldn't contain '.', for oracle
                        String minMax = resultSet.getString(1) + resultSet.getString(2);
                        if (StringUtils.contains(minMax, '.'))
                        {
                            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK,
                                    "您配置的DataX切分主键(splitPk)有误. 因为您配置的切分主键(splitPk) 类型 DataX 不支持. DataX 仅支持切分主键为一个,并且类型为整数或者字符串类型. 请尝试使用其他的切分主键或者联系 DBA 进行处理.");
                        }
                    }
                } else
                {
                    throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK,
                            "您配置的DataX切分主键(splitPk)有误. 因为您配置的切分主键(splitPk) 类型 DataX 不支持. DataX 仅支持切分主键为一个,并且类型为整数或者字符串类型. 请尝试使用其他的切分主键或者联系 DBA 进行处理.");
                }
            } else
            {
                throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK,
                        "您配置的DataX切分主键(splitPk)有误. 因为您配置的切分主键(splitPk) 类型 DataX 不支持. DataX 仅支持切分主键为一个,并且类型为整数或者字符串类型. 请尝试使用其他的切分主键或者联系 DBA 进行处理.");
            }
        } catch (DataExchangeException e)
        {
            throw e;
        } catch (Exception e)
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK, "DataX尝试切分表发生错误. 请检查您的配置并作出修改.", e);
        } finally
        {
            Jdbc.close(resultSet, null, null);
        }

        return minMaxPK;
    }

    private static boolean isPKTypeValid(ResultSetMetaData resultSetMetaData)
    {
        boolean ret = false;
        try
        {
            int minType = resultSetMetaData.getColumnType(1);
            int maxType = resultSetMetaData.getColumnType(2);

            boolean isNumberType = isLongType(minType);
            boolean isStringType = isStringType(minType);

            if (minType == maxType && (isNumberType || isStringType))
            {
                ret = true;
            }
        } catch (SQLException e)
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_SPLIT_PK,
                    "DataX获取切分主键(splitPk)字段类型失败. 该错误通常是系统底层异常导致. 请联系旺旺:askdatax或者DBA处理.");
        }
        return ret;
    }

    // warn: Types.NUMERIC is used for oracle! because oracle use NUMBER to
    // store INT, SMALLINT, INTEGER etc, and only oracle need to concern
    // Types.NUMERIC
    private static boolean isLongType(int type)
    {
        boolean isValidLongType = type == Types.BIGINT || type == Types.INTEGER
                || type == Types.SMALLINT || type == Types.TINYINT;


        return isValidLongType;
    }

    private static boolean isStringType(int type)
    {
        return type == Types.CHAR || type == Types.NCHAR
                || type == Types.VARCHAR || type == Types.LONGVARCHAR
                || type == Types.NVARCHAR;
    }

    private static String genPKRangeSQL(Configuration configuration)
    {

        String splitPK = configuration.getString(Key.split_key).trim();
        String table = configuration.getString(Key.table_name).trim();
        String where = configuration.getString(Key.where, null);
        return genPKSql(splitPK, table, where);
    }

    public static String genPKSql(String splitPK, String table, String where)
    {

        String minMaxTemplate = "SELECT MIN(%s),MAX(%s) FROM %s";
        String pkRangeSQL = String.format(minMaxTemplate, splitPK, splitPK,
                table);
        if (StringUtils.isNotBlank(where))
        {
            pkRangeSQL = String.format("%s where (%s AND %s IS NOT NULL)",
                    pkRangeSQL, where, splitPK);
        }
        return pkRangeSQL;
    }


}